package com.shux.kafka.producer;

import java.util.Properties;
import java.util.concurrent.Future;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;


/**
 * 描述：
 * 
 * <pre>HISTORY
 * ****************************************************************************
 *  ID   DATE           PERSON          REASON
 *  1    ${date}      Simba.Hua         Create
 * ****************************************************************************
 * </pre>
 * @author Simba.Hua
 */
public  class KafkaProducerServer<K,V> {
	private ProducerConfig<K,V> config;
	private Producer<K,V> producer;
	public KafkaProducerServer(ProducerConfig<K,V> config){
		this.config = config;
		producer = new KafkaProducer<>(initProperties(),config.getKeySerializer(),config.getValueSerializer());
	}
	public Future<RecordMetadata> sendMessage(V message) throws Exception{
		try {
			return producer.send(new ProducerRecord<K, V>(config.getTopic(),message));
			
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}finally{
			//producer.close();
		}
		return null;
	}
	private Properties initProperties(){
		 Properties props = new Properties();
		 //10.202.34.28:2182,10.202.34.29:2182,10.202.34.30:2182/kafka/other
		 props.put("bootstrap.servers", config.getKafkaServer());
		 props.put("acks", config.getAcks());
		 props.put("retries", config.getRetries());
		 props.put("batch.size", config.getBatchSize());
		 props.put("linger.ms", config.getLingerMs());
		 props.put("buffer.memory", config.getBufferMemory());
		 props.put("key.serializer", config.getKeySerializer().getClass().getName());
		 props.put("value.serializer", config.getValueSerializer().getClass().getName());
		 return props;
	}
}
